package io.kiki.sba.registry.remoting.netty;


import io.kiki.sba.registry.api.*;
import io.kiki.sba.registry.api.ChannelHandler.HandlerType;
import io.kiki.sba.registry.api.ChannelHandler.InvokeType;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.net.NetUtil;
import io.kiki.stack.netty.channelManager.ChannelEventProcessor;
import io.kiki.stack.netty.channelManager.ChannelEventType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;


public class NettyClient implements Client {

    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    private final io.kiki.stack.netty.Client client;

    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final int connNum;
    private int connectTimeout = 2000;


    public NettyClient(int connNum) {
        client = new io.kiki.stack.netty.Client();
        configIO();
        client.init();

        this.connNum = connNum;
    }

    private void configIO() {
        final int low = Integer.getInteger(io.kiki.stack.netty.config.Configs.NETTY_BUFFER_LOW_WATERMARK, 1024 * 256);
        final int high = Integer.getInteger(io.kiki.stack.netty.config.Configs.NETTY_BUFFER_HIGH_WATERMARK, 1024 * 288);
        client.initWriteBufferWaterMark(low, high);
        logger.info("config watermark, low={}, high={}", low, high);
    }

    public Map<String, List<io.kiki.stack.netty.channelManager.Channel>> getConnections() {
        return client.getAllManagedChannels();
    }


    public void initHandlers(List<ChannelHandler> channelHandlerList) {
        final ChannelHandler channelhandler1 = BoltUtil.getListenerHandlers(channelHandlerList);

        client.addChannelEventProcessor(ChannelEventType.CONNECT, newConnectionEventAdapter(channelhandler1, ChannelEventType.CONNECT));
        client.addChannelEventProcessor(ChannelEventType.CLOSE, newConnectionEventAdapter(channelhandler1, ChannelEventType.CLOSE));
        client.addChannelEventProcessor(ChannelEventType.EXCEPTION, newConnectionEventAdapter(channelhandler1, ChannelEventType.EXCEPTION));

        for (ChannelHandler channelHandler : channelHandlerList) {
            if (HandlerType.PROCESSER.equals(channelHandler.getHandlerType())) {
                if (InvokeType.SYNC.equals(channelHandler.getInvokeType())) {
                    client.registerUserProcessor(newSyncProcessor(channelHandler));
                } else {
                    client.registerUserProcessor(newAsyncProcessor(channelHandler));
                }
            }
        }
    }

    protected ChannelEventProcessor newConnectionEventAdapter(ChannelHandler channelHandler, ChannelEventType channelEventType) {
        return new ChannelEventProcessorImpl(channelEventType, channelHandler);
    }

    protected AsyncUserProcessorAdapter newAsyncProcessor(ChannelHandler channelHandler) {
        return new AsyncUserProcessorAdapter(channelHandler);
    }

    protected SyncUserProcessorAdapter newSyncProcessor(ChannelHandler channelHandler) {
        return new SyncUserProcessorAdapter(channelHandler);
    }

    @Override
    public io.kiki.sba.registry.api.Channel connect(Url url) {
        if (url == null) {
            throw new IllegalArgumentException("Create connection url can not be null!");
        }
        try {
            io.kiki.stack.netty.channelManager.Channel channel = getBoltConnection(client, url);
            return new Channel(channel);
        } catch (RemotingException e) {
            throw BoltUtil.handleException("BoltClient", url, e, "connect");
        }
    }

    protected io.kiki.stack.netty.channelManager.Channel getBoltConnection(io.kiki.stack.netty.Client client, Url url) throws RemotingException {
        io.kiki.stack.netty.channelManager.Url url1 = createUrl(url);
        try {
            io.kiki.stack.netty.channelManager.Channel channel = client.getChannel(url1, connectTimeout);
            if (channel == null || !channel.isFine()) {
                if (channel != null) {
                    channel.close();
                }
                throw new ChannelConnectException("Get bolt connection failed for boltUrl: " + url1);
            }
            return channel;
        } catch (InterruptedException e) {
            throw BoltUtil.handleException("BoltClient", url1, e, "getConnection");
        } catch (io.kiki.stack.netty.exception.RemotingException e) {
            throw new RuntimeException(e);
        }
    }

    protected io.kiki.stack.netty.channelManager.Url createUrl(Url url) {
        io.kiki.stack.netty.channelManager.Url url1 = new io.kiki.stack.netty.channelManager.Url(url.getIp(), url.getPort());

        url1.setChannelNum(connNum);
        url1.setConnWarmup(true);
        return url1;
    }

    @Override
    public io.kiki.sba.registry.api.Channel getChannel(Url url) {
        try {
            io.kiki.stack.netty.channelManager.Channel channel1 = getBoltConnection(client, url);
            Channel channel = new Channel(channel1);
            return channel;
        } catch (Throwable e) {
            throw BoltUtil.handleException("BoltClient", url, e, "getChannel");
        }
    }

    @Override
    public InetSocketAddress getLocalAddress() {
        return NetUtil.getLocalSocketAddress();
    }

    @Override
    public void close() {
        if (closed.compareAndSet(false, true)) {
            client.shutdown();
        }
    }

    @Override
    public boolean isClosed() {
        return closed.get();
    }

    @Override
    public Object sendSync(Url url, Object message, int timeoutMillis) {
        try {
            io.kiki.stack.netty.channelManager.Url url1 = createUrl(url);
            return client.invokeSync(url1, message, timeoutMillis);
        } catch (Throwable e) {
            throw BoltUtil.handleException("BoltClient", url, e, "sendSync");
        }
    }

    @Override
    public Object sendSync(io.kiki.sba.registry.api.Channel channel, Object message, int timeoutMillis) {
        BoltUtil.checkChannelConnected(channel);
        try {
            return client.invokeSync(((Channel) channel).getChannel(), message, timeoutMillis);
        } catch (Throwable e) {
            throw BoltUtil.handleException("BoltClient", channel, e, "sendSync");
        }
    }

    @Override
    public void sendWithCallback(Url url, Object message, Callback callback, int timeoutMillis) {
        try {
            io.kiki.stack.netty.channelManager.Channel connection = getBoltConnection(client, url);
            client.invokeWithCallback(connection, message, new InvokeCallbackHandler(new Channel(connection), callback), timeoutMillis);
            return;
        } catch (RemotingException e) {
            throw BoltUtil.handleException("BoltClient", url, e, "sendCallback");
        } catch (io.kiki.stack.netty.exception.RemotingException e) {
            throw new RuntimeException(e);
        }
    }
}
